package org.osami.gateway.mediator.jms.collector;

import javax.jms.BytesMessage;
import javax.jms.Connection;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.ObjectMessage;
import javax.jms.Queue;
import javax.jms.QueueConnectionFactory;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.naming.InitialContext;

import org.osgi.service.log.LogService;

import fr.liglab.adele.cilia.Data;
import fr.liglab.adele.cilia.framework.AbstractCollector;

public class JmsQueueConsumerCollector extends AbstractCollector implements
		MessageListener {
	
    private String m_instanceName;

    // The JNDI names of the JMS resources needed by this mediator
    
    private String m_queueConnectionFactoryName;
    private String m_queueName;

    // The needed resources retrieved via JNDI
    private volatile QueueConnectionFactory m_queueConnectionFactory;
    private volatile Queue m_queue;
    private volatile Connection conn;
    private volatile Session sess;

    // Logger
    private LogService m_logger;

	public synchronized void start(){
		// Need to switch TCCL for JNDI lookup
        ClassLoader old = Thread.currentThread().getContextClassLoader();
        try {
            // Replace the TCCL with this bundle class loader (dynamic imports
            // must be enabled)
            Thread.currentThread().setContextClassLoader(
                    JmsQueueConsumerCollector.class.getClassLoader());
            // Get JNDI initial context
            InitialContext context;
            try {
                context = new InitialContext();
            } catch (Exception e) {
                log(LogService.LOG_ERROR, "Cannot get JNDI initial context", e);
                throw new RuntimeException(e);
            }
            // Get JMS QueueConnectionFactory
            try {
                m_queueConnectionFactory = (QueueConnectionFactory) context
                        .lookup(m_queueConnectionFactoryName);
            } catch (Exception e) {
                log(LogService.LOG_ERROR,
                        "Cannot get JMS QueueConnectionFactory", e);
                throw new RuntimeException(e);
            }
            // Get JMS Queue
            try {
                m_queue = (Queue) context.lookup(m_queueName);
            } catch (Exception e) {
                log(LogService.LOG_ERROR, "Cannot get JMS Queue", e);
                throw new RuntimeException(e);
            }
            
            try {
				conn = m_queueConnectionFactory.createConnection();
				sess = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
				MessageConsumer mc = sess.createConsumer(m_queue);
				mc.setMessageListener(this);
				conn.start();
			} catch (JMSException e) {
				log(LogService.LOG_ERROR, "Cannot start the connection",e);
			}
            
            // Log success
            log(LogService.LOG_INFO, "Started");
            log(LogService.LOG_DEBUG, "queueConnectionFactory.name = "
                    + m_queueConnectionFactoryName);
            log(LogService.LOG_DEBUG, "queue.name = " + m_queueName);
            log(LogService.LOG_INFO, "Consumer Connection started & listener registered");
        } finally {
            // Restore TCCL
            Thread.currentThread().setContextClassLoader(old);
        }
		
	}
	
	public synchronized void stop(){
		m_queueConnectionFactory=null;
		m_queue=null;
		try {
			sess.close();
			conn.close();
		} catch (JMSException e) {
			log(LogService.LOG_ERROR,"Couldn't close connection",e);
		}
		log(LogService.LOG_INFO,"Stopped");
	}
	
	public synchronized void onMessage(Message msg) {
		try {
		      if(msg==null){
		        log(LogService.LOG_WARNING, "Null Message received");
		      }
		      else {
		    	  Data d=null;
		        if(msg instanceof ObjectMessage) {
		            Object o = ((ObjectMessage)msg).getObject();
		            d = new Data(o);
		            log(LogService.LOG_INFO, "Object message received : "+o);
		        } else if(msg instanceof TextMessage) {
		            String m = ((TextMessage)msg).getText();
		            d = new Data(m);
		            log(LogService.LOG_INFO, "Text message received : "+m);
		        } else if(msg instanceof BytesMessage) {
		        	//TODO
		        	//Extract bytes array and create a data object
		        	log(LogService.LOG_INFO, "Bytes message received with size : "+((BytesMessage)msg).getBodyLength());
		        }
		        this.notifyDataArrival(d);
		      }
		      }catch(Exception e) {
		          log(LogService.LOG_ERROR,"Exception during message receiving"+e);
		          throw new RuntimeException(e);
		      }
	}
	
    // Utility methods

    private void log(int level, String message, Throwable throwable) {
        m_logger.log(level, m_instanceName + " - " + message, throwable);
    }

    private void log(int level, String message) {
        log(level, message, null);
    }

}
